package cn.item.buckle.business;

import cn.item.buckle.business.pojo.MonitorCameraState;
import org.apache.spark.util.AccumulatorV2;

import java.util.HashMap;

public class MonitorAccumulator extends AccumulatorV2<MonitorCameraState, MonitorCameraState> {
	private final MonitorCameraState mcs = new MonitorCameraState(0, 0, 0, 0, new HashMap<>());

	@Override
	public boolean isZero() {
		//判断，这个要和reset设定值一致
		return mcs.getNormalMonitorCount() == 0 && mcs.getNormalCameraCount() == 0 &&
				mcs.getAbnormalMonitorCount() == 0 && mcs.getAbnormalCameraCount() == 0
				&& mcs.getAbnormalMonitorCameraInfo().isEmpty();
	}

	@Override
	public AccumulatorV2<MonitorCameraState, MonitorCameraState> copy() {
		//复制一个新对象
		return new MonitorAccumulator();
	}

	@Override
	public void reset() {
		//重置每个分区的数值
		mcs.setNormalMonitorCount(0);
		mcs.setAbnormalCameraCount(0);
		mcs.setAbnormalMonitorCount(0);
		mcs.setAbnormalCameraCount(0);
		mcs.setAbnormalMonitorCameraInfo(new HashMap<>());
	}

	@Override
	public void add(MonitorCameraState monitorCameraState) {
		//每个分区累加自己的数值
		mcs.addNormalMonitorCount(monitorCameraState.getNormalMonitorCount());
		mcs.addNormalCameraCount(monitorCameraState.getNormalCameraCount());
		mcs.addAbnormalMonitorCount(monitorCameraState.getAbnormalMonitorCount());
		mcs.addAbnormalCameraCount(monitorCameraState.getAbnormalCameraCount());
		mcs.addAbnormalMonitorCameraInfo(monitorCameraState.getAbnormalMonitorCameraInfoForMap());
	}

	@Override
	public void merge(AccumulatorV2<MonitorCameraState, MonitorCameraState> accumulatorV2) {
		//合并分区值，求得总值
		this.add(accumulatorV2.value());
	}

	@Override
	public MonitorCameraState value() {
		//返回结果
		return this.mcs;
	}

}
